package producer_test

import (
	"container/list"
	"context"
	"fmt"
	"gitee.com/zaper/kafka-service/model"
	"gitee.com/zaper/kafka-service/service/connect"
	metadata "gitee.com/zaper/kafka-service/service/metadata"
	"gitee.com/zaper/kafka-service/util"
	"gitee.com/zaper/private-logger"
	"github.com/petermattis/goid"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/protocol"
	"math/rand"
	"strconv"
	"time"
)

// ExecClientProductMess 使用client.Produce的生产者模型
func ExecClientProductMess() {
	conn := connect.GetKafkaConnectManager()
	ctx := context.Background()
	// 创建消息记录
	var topicName = "TestTopic"
	var partitions = metadata.GetMetadataManager().GetAllPartitionsByTopicName(topicName)
	records := make([]protocol.Record, 0)
	for i := 0; i < 50; i++ {
		record := protocol.Record{
			Key:   protocol.NewBytes([]byte("key1")),
			Value: protocol.NewBytes([]byte("world " + strconv.Itoa(i))),
		}
		records = records[:0]
		records = append(records, record)
		resp, _ := conn.Client.Produce(ctx, &kafka.ProduceRequest{
			Topic:     topicName,
			Partition: rand.Intn(len(partitions)),
			Records:   kafka.NewRecordReader(records...),
		})
		time.Sleep(100 * time.Millisecond)
		fmt.Println(resp)
	}
}

// execSegmentProducer 使用Writer的生产者模型
func execSegmentProducer(topList *list.List, producerModel *model.KafkaProducerModel) {
	// 每个生产者者先轮训后随机获取一个主题(当轮旋完所有主题之后保证所有主题都被消费者占用就开始随机)
	topicName := cusTopicName(topList, producerModel)
	// 创建Kafka写入器(最简配置，优化项配置参考tmp目录producer的模型)
	w := &kafka.Writer{
		Addr:         kafka.TCP([]string{"localhost:9091"}...),
		Topic:        topicName,
		BatchSize:    1, // 关键配置：每条消息立即发送
		Balancer:     &kafka.LeastBytes{},
		RequiredAcks: kafka.RequireAll,
		ReadTimeout:  10 * time.Second,
		WriteTimeout: 10 * time.Second,
	}
	defer util.CloseWRIo(&util.KafkaWRIo{Writer: w})
	defer close(producerModel.StopChan[goid.Get()])
	ctx := context.Background()
	// 消费者消息表示
	i := 0
	for {
		select {
		case <-producerModel.StopChan[goid.Get()]:
			logger.Info("kafka segment producer shutting down")
			deleteSignMap(producerModel.StopChan, goid.Get())
			return
		case <-time.After(2 * time.Second):
		}
		value := fmt.Sprintf("%d号生产者使用segmentio生产器生产数据=%s", goid.Get(), strconv.Itoa(i))
		message := &kafka.Message{
			Value: []byte(value),
		}
		// 发送消息
		err := w.WriteMessages(ctx, *message) // 忽略了err，如果err了消费者可能收不到消息
		if err != nil {
			logger.Infof("发送消息失败: %v", err)
			continue
		}
		logger.Infof("生产者：%d号生产者使用segmentio生产器生产数据=%s，发送消息到主题%s",
			goid.Get(), strconv.Itoa(i), topicName)
		i++
	}
}
